// 修改后的 RegionOrderMapper.java
package com.taxi.analysis.regionTop10;

import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class RegionOrderMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    private final static IntWritable one = new IntWritable(1);
    private Text region = new Text();

    @Override
    public void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {

        // 跳过标题行
        if (key.get() == 0) {
            return;
        }

        try {
            String line = value.toString().trim();
            if (line.isEmpty()) {
                return;
            }

            // 解析CSV行 - 使用更健壮的解析方法
            String[] fields = parseCSVLine(line);

            // 打印调试信息（前几行）
            if (key.get() < 1000) {
                System.err.println("Debug - Line: " + line);
                System.err.println("Debug - Fields length: " + fields.length);
                if (fields.length > 2) {
                    System.err.println("Debug - Field[2] (pickup_zone): " + fields[2]);
                }
            }

            // 根据实际的CSV结构调整字段索引
            if (fields.length >= 3) {
                String pickupZone = fields[2].trim(); // pickup_zone字段

                // 过滤无效数据
                if (!pickupZone.isEmpty() &&
                        !pickupZone.equals("null") &&
                        !pickupZone.equals("pickup_zone") &&
                        !pickupZone.matches("^\\d+\\.?\\d*$")) { // 排除纯数字

                    region.set(pickupZone);
                    context.write(region, one);
                }
            }
        } catch (Exception e) {
            System.err.println("Error processing line: " + value.toString());
            e.printStackTrace();
        }
    }

    /**
     * 更健壮的CSV解析方法
     */
    private String[] parseCSVLine(String line) {
        // 处理可能包含引号的CSV
        if (line.contains("\"")) {
            return parseQuotedCSV(line);
        } else {
            return line.split(",");
        }
    }

    private String[] parseQuotedCSV(String line) {
        // 简单的引号处理
        java.util.List<String> fields = new java.util.ArrayList<>();
        boolean inQuotes = false;
        StringBuilder currentField = new StringBuilder();

        for (int i = 0; i < line.length(); i++) {
            char c = line.charAt(i);

            if (c == '"') {
                inQuotes = !inQuotes;
            } else if (c == ',' && !inQuotes) {
                fields.add(currentField.toString());
                currentField = new StringBuilder();
            } else {
                currentField.append(c);
            }
        }
        fields.add(currentField.toString());

        return fields.toArray(new String[0]);
    }
}
